package cloud.xiguapi.ubas.common;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Flink Environment
 *
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-16 下午 11:14
 * desc:
 */
public class FlinkEnvironment {

    /**
     * 默认并行度为8
     */
    public static final int DEFAULT_PARALLELISM = 8;
    /**
     * 默认时间语义为事件时间
     */
    public static final TimeCharacteristic DEFAULT_CHARACTERISTIC = TimeCharacteristic.EventTime;

    private FlinkEnvironment() {
    }

    /**
     * 获取默认的流式执行环境
     * - 默认并行度: 8
     * - 默认时间语义: 事件时间
     *
     * @return 默认流式执行环境
     */
    public static StreamExecutionEnvironment defaultEnvironment() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(DEFAULT_PARALLELISM);
        env.setStreamTimeCharacteristic(DEFAULT_CHARACTERISTIC);
        return env;
    }

    /**
     * 获取原始流式执行环境
     *
     * @return 原始流式执行环境
     */
    public static StreamExecutionEnvironment environment() {
        return StreamExecutionEnvironment.getExecutionEnvironment();
    }

    /**
     * 获取流式执行环境
     *
     * @param parallelism 并行度
     * @return 流式执行环境
     */
    public static StreamExecutionEnvironment environment(int parallelism) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(parallelism);
        return env;
    }

    /**
     * 获取流式执行环境
     *
     * @param parallelism    并行度
     * @param characteristic 时间语义
     * @return 流式执行环境
     */
    public static StreamExecutionEnvironment environment(int parallelism, TimeCharacteristic characteristic) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(parallelism);
        env.setStreamTimeCharacteristic(characteristic);
        return env;
    }

    /**
     * 获取批处理式执行环境
     *
     * @return 批处理执行环境
     */
    public static ExecutionEnvironment datasetEnvironment() {
        return ExecutionEnvironment.getExecutionEnvironment();
    }
}
